package bsp;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.io.Writable;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

public abstract class BspHelper<IK, IV, OK, OV, MS extends Writable> extends
		BSP<IK, IV, OK, OV, MS> {

	protected ArrayList<CMapper<IK, IV, OK, OV, MS>> mappers;
	protected ArrayList<CReducer<IK, IV, OK, OV, MS>> reducers;

	public BspHelper() {
		mappers = new ArrayList<CMapper<IK, IV, OK, OV, MS>>();
		reducers = new ArrayList<CReducer<IK, IV, OK, OV, MS>>();
	}

	public void addMapper(CMapper<IK, IV, OK, OV, MS> mapper) {
		// System.out.println("*****set a map*****");
		if (mappers == null)
			mappers = new ArrayList<CMapper<IK, IV, OK, OV, MS>>();
		mappers.add(mapper);
	}

	public void addReducer(CReducer<IK, IV, OK, OV, MS> reducer) {
		reducers.add(reducer) ;
	}
	@Override
	public void bsp(BSPPeer<IK, IV, OK, OV, MS> peer) throws IOException,
			SyncException, InterruptedException {
		int count = 0;
		for (CMapper<IK, IV, OK, OV, MS> mapper : mappers) {
			System.out.println("*****BspHelper::map*****" + count);
			mapper.setup(peer,null);
			mapper.mapAll(peer);
			// add to cache
			mapper.sendAllMesgs(peer);
			System.out.println("*****BspHelper::send*****");
			peer.reopenInput();
			//update cache and output 
			count ++;
			peer.clear(); //TODO just for test
			peer.sync();
		}
		
		//if(reducers != null)
		for (CReducer<IK, IV, OK, OV, MS> reducer : reducers){
			reducer.setup(peer);
			reducer.reduce(peer);
			reducer.sendAllMesgs(peer);
			peer.sync();
		}

	}

	public abstract void setAllCMappers();

	public abstract void setAllCReducers();

	@Override
	public void setup(BSPPeer<IK, IV, OK, OV, MS> peer) throws IOException,
			SyncException, InterruptedException {
		setAllCMappers();
		setAllCReducers();
		//set cache
	}

}
